package com.dzsw.rabbitmq.mq;

import com.dzsw.rabbitmq.config.RabbitConfig;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * 需要注意的 basicAck 方法需要传递两个参数
 *
 * deliveryTag（唯一标识 ID）：当一个消费者向 RabbitMQ 注册后，会建立起一个 Channel ，RabbitMQ 会用 basic.deliver 方法向消费者推送
 * 消息，这个方法携带了一个 delivery tag， 它代表了 RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID，是一个单调递增的正整数，
 * delivery tag的范围仅限于 Channel
 * multiple：为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
 *  
 *
 * basicNack方法需要传递三个参数
 *
 * deliveryTag（唯一标识 ID）：上面已经解释了。
 * multiple：上面已经解释了。
 * requeue： true ：重回队列，false ：丢弃，我们在nack方法中必须设置 false，否则重发没有意义。
 *  
 *
 * basicReject方法需要传递两个参数
 *
 * deliveryTag（唯一标识 ID）：上面已经解释了。
 * requeue：上面已经解释了，在reject方法里必须设置true。
 */
@Component
@Slf4j
public class MsgConsumer {

    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue(value = RabbitConfig.FANOUT_QUEUE_NAME, durable = "true"),
                    exchange = @Exchange(value = RabbitConfig.TEST_FANOUT_EXCHANGE, type = "fanout"))
    })
    @RabbitHandler
    public void processFanoutMsg(Message message, Channel channel, @Headers Map<String, Object> headers) throws Exception {
        try {
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            log.info("received Fanout message : " + msg);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            log.info(String.format("消费消息确认[%s]，接收到了回调方法", message.getMessageProperties().getConsumerQueue()));
        } catch (Exception ex) {
            //重新回到队列
//            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
//            System.out.println("尝试重发：" + message.getMessageProperties().getConsumerQueue());
            //requeue =true 重回队列，false 丢弃
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            // TODO 该消息已经导致异常，重发无意义，自己实现补偿机制
        }
    }

    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue(value = RabbitConfig.FANOUT_QUEUE_NAME1, durable = "true"),
                    exchange = @Exchange(value = RabbitConfig.TEST_FANOUT_EXCHANGE, type = "fanout"))
    })
    @RabbitHandler
    public void processFanout1Msg(Message message, Channel channel, @Headers Map<String, Object> headers) throws Exception {
        try {
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            log.info("received Fanout1 message : " + msg);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            log.info(String.format("消费消息确认[%s]，接收到了回调方法", message.getMessageProperties().getConsumerQueue()));
        } catch (Exception ex) {
            //重新回到队列
//            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
//            System.out.println("尝试重发：" + message.getMessageProperties().getConsumerQueue());
            //requeue =true 重回队列，false 丢弃
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            // TODO 该消息已经导致异常，重发无意义，自己实现补偿机制
        }
    }

    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue(value = RabbitConfig.DIRECT_QUEUE_NAME, durable = "true"),
                    exchange = @Exchange(value = RabbitConfig.TEST_DIRECT_EXCHANGE),
                    key = RabbitConfig.DIRECT_ROUTINGKEY)
    })
    @RabbitHandler
    public void processDirectMsg(Message message, Channel channel, @Headers Map<String, Object> headers) throws Exception {
        try {
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            log.info("received Fanout1 message : " + msg);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            log.info(String.format("消费消息确认[%s]，接收到了回调方法", message.getMessageProperties().getConsumerQueue()));
        } catch (Exception ex) {
            //重新回到队列
//            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
//            System.out.println("尝试重发：" + message.getMessageProperties().getConsumerQueue());
            //requeue =true 重回队列，false 丢弃
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            // TODO 该消息已经导致异常，重发无意义，自己实现补偿机制
        }
    }

    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue(value = RabbitConfig.TOPIC_QUEUE_NAME, durable = "true"),
                    exchange = @Exchange(value = RabbitConfig.TEST_TOPIC_EXCHANGE, type = "topic"),
                    key = RabbitConfig.TOPIC_ROUTINGKEY)
    })
    @RabbitHandler
    public void processTopicMsg(Message message, Channel channel, @Headers Map<String, Object> headers) throws Exception {
        try {
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            log.info("received Topic message : " + msg);
            channel.basicAck(deliveryTag, true);
            log.info(String.format("消费消息确认[%s]，接收到了回调方法", message.getMessageProperties().getConsumerQueue()));
        } catch (Exception ex) {
            //重新回到队列
//            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
//            System.out.println("尝试重发：" + message.getMessageProperties().getConsumerQueue());
            //requeue =true 重回队列，false 丢弃
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            // TODO 该消息已经导致异常，重发无意义，自己实现补偿机制
        }
    }


}
